#include "requestor.hpp"

namespace yjz_rpc
{
    namespace client
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            using SubCallback = std::function<void(const std::string& key, const std::string& msg)>;

            TopicManager(const Requestor::ptr& requestor) : _requestor(requestor) {}

            bool create(const BaseConnection::ptr& conn, const std::string& key)
            {
                return commonRequest(conn, key, TopicOpType::TOPIC_CREATE);
            }

            bool remove(const BaseConnection::ptr& conn, const std::string& key)
            {
                return commonRequest(conn, key, TopicOpType::TOPIC_REMOVE);
            }

            bool subscribe(const BaseConnection::ptr& conn, const std::string& key, const SubCallback& cb)
            {
                addSubscribe(key, cb);
                bool res = commonRequest(conn, key, TopicOpType::TOPIC_SUBSCRIBE);
                if (res == false)
                {
                    delSubscribe(key);
                    return false;
                }
                return true;
            }

            bool publish(const BaseConnection::ptr& conn, const std::string& key, const std::string& msg)
            {
                return commonRequest(conn, key, TopicOpType::TOPIC_PUBLISH, msg);
            }

            bool cancel(const BaseConnection::ptr& conn, const std::string& key)
            {
                delSubscribe(key);
                return commonRequest(conn, key, TopicOpType::TOPIC_CANCEL);
            }

            void onPublish(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg)
            {
                auto type = msg->opType();
                // 客户端只能处理服务端推送过来的消息
                if (type != TopicOpType::TOPIC_PUBLISH)
                {
                    LOG(LogLevel::ERROR) << "主题操作类型错误！";
                    return;
                }
                std::string topic_key = msg->topicKey();
                std::string topic_msg = msg->topicMsg();
                // 通过主题名称查找对应的回调处理函数
                auto callback = getSubscribe(topic_key);
                if (!callback)
                {
                    LOG(LogLevel::ERROR) << "收到的消息没有对应的回调处理函数！";
                    return;
                }
                return callback(topic_key, topic_msg);
            }
        private:
            void addSubscribe(const std::string& key, const SubCallback& cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.insert(std::make_pair(key, cb));
            }

            void delSubscribe(const std::string& key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.erase(key);
            }

            const SubCallback getSubscribe(const std::string& key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_callbacks.find(key);
                if (it == _topic_callbacks.end())
                {
                    return SubCallback();
                }
                return it->second;
            }

            bool commonRequest(const BaseConnection::ptr& conn, const std::string& key, TopicOpType type, 
                const std::string& msg = "")
            {
                // 1、构建请求对象
                auto req_msg = MessageFactory::create<TopicRequest>();
                req_msg->setId(UUID::uuid());
                req_msg->setMType(MType::REQ_TOPIC);
                req_msg->setOpType(type);
                req_msg->setTopicKey(key);
                if (type == TopicOpType::TOPIC_PUBLISH)
                {
                    req_msg->setTopicMsg(msg);
                }

                // 2、向服务器发送请求，等待响应
                BaseMessage::ptr rsp_msg;
                bool res = _requestor->send(conn, req_msg, rsp_msg);
                if (res == false)
                {
                    LOG(LogLevel::ERROR) << "主题操作请求失败！";
                    return false;
                }
                
                // 3、判断请求处理是否成功
                auto topic_rsp_msg = std::dynamic_pointer_cast<TopicResponse>(rsp_msg);
                if (topic_rsp_msg.get() == nullptr)
                {
                    LOG(LogLevel::ERROR) << "主题操作响应向下类型转换失败！";
                    return false;
                }
                if (topic_rsp_msg->rcode() != RCode::RCODE_OK)
                {
                    LOG(LogLevel::ERROR) << "主题操作请求出错： " << errReason(topic_rsp_msg->rcode());
                    return false;
                }
                return true;
            }
        private:
            std::mutex _mutex;
            Requestor::ptr _requestor;
            std::unordered_map<std::string, SubCallback> _topic_callbacks;
        };
    }
}